package com.atguigu.realtime.util;

import com.atguigu.realtime.annotation.NoSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @ClassName: MySinkUtil
 * @Description:
 * @Author: kele
 * @Date: 2021/4/25 15:39
 **/
public class MySinkUtil {

    public static <T> SinkFunction<T> getClickHouseSink(String db,
                                                        String table,
                                                        Class<T> tClass) {

        String url = "jdbc:clickhouse://hadoop162:8123/" + db;
        String driver = "ru.yandex.clickhouse.ClickHouseDriver";
        StringBuilder sql = new StringBuilder();

        // insert into user (id, ....) values (?, ...)
        sql.append("insert into ").append(table).append(" (");

        Field[] fields = tClass.getDeclaredFields();

        for (Field field : fields) {
            if(field.getAnnotation(NoSink.class) == null){
                //添加了NoSink的字段不参与字段名
                sql.append(field.getName()).append(",");
            }
        }
        sql.deleteCharAt(sql.length()-1);

        sql.append(") values ( ");

        for (Field field : fields) {

            if (field.getAnnotation(NoSink.class) == null) {
                //添加了NoSink的字段不参与字段名
                sql.append("?,");
            }
        }
        sql.deleteCharAt(sql.length()-1);
        sql.append(")");

        return returnSinkFunction(url,driver,sql.toString());
    }

    public static <T> SinkFunction<T> returnSinkFunction(String url,
                                                        String driver,
                                                        String sql) {

       return  JdbcSink.sink(sql,

                new JdbcStatementBuilder<T>(){

                    /**
                     * 该方法是给占位符进行赋值
                     * @param ps
                     * @param t
                     * @throws SQLException
                     */
                    @Override
                    public void accept(PreparedStatement ps,
                                       T t) throws SQLException {

                        Field[] fields = t.getClass().getDeclaredFields();

                        for (int i = 0,j = 1; i < fields.length; i++) {

                            Field field = fields[i];

                            if(field.getAnnotation(NoSink.class) == null){
                                //通过反射取出字段中的值
                                try {
                                    //暴力破解，得到对象中的属性值
                                    field.setAccessible(true);
                                    //取出
                                    Object value = field.get(t);

                                    ps.setObject(j++,value);

                                } catch (IllegalAccessException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                },

               //设置批次
               new JdbcExecutionOptions.Builder()
                       .withBatchSize(1)
                       .build(),

                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .build()
        );
    }

}





